package com.nehc.nettyserver.netty;

import com.nehc.nettyserver.common.model.ClientInfo;
import com.nehc.nettyserver.service.cache.CacheService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.util.Objects;
import java.util.function.Function;
import java.util.HashSet;

/**
 * 通道管理器
 * 负责管理所有连接的Channel
 * 
 * @author NEHC
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ChannelManager {
    
    public static final AttributeKey<String> ATTR_CLIENT_ID = AttributeKey.valueOf("clientId");
    public static final AttributeKey<String> ATTR_USER_ID = AttributeKey.valueOf("userId");
    public static final AttributeKey<Boolean> ATTR_AUTHORIZED = AttributeKey.valueOf("authorized");
    
    private final CacheService cacheService;
    
    private final ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private final Map<String, ChannelId> clientIdChannelMap = new ConcurrentHashMap<>();
    private final Map<String, Set<String>> userIdClientIdsMap = new ConcurrentHashMap<>();

    /**
     * 注册新连接
     * 
     * @param channel 通道
     * @param clientId 客户端ID
     */
    public void register(Channel channel, String clientId) {
        channel.attr(ATTR_CLIENT_ID).set(clientId);
        channel.attr(ATTR_AUTHORIZED).set(false);
        
        channelGroup.add(channel);
        clientIdChannelMap.put(clientId, channel.id());
        
        // 记录连接到缓存
        ClientInfo clientInfo = new ClientInfo();
        clientInfo.setClientId(clientId);
        clientInfo.setConnectTime(System.currentTimeMillis());
        clientInfo.setRemoteAddress(channel.remoteAddress().toString());
        cacheService.put(CacheService.CacheType.CONNECTION, clientId, clientInfo);
        
        log.info("客户端[{}]已连接，当前连接数: {}", clientId, channelGroup.size());
    }
    
    /**
     * 授权通道
     * 
     * @param channel 通道
     * @param userId 用户ID
     */
    public void authorize(Channel channel, String userId) {
        String clientId = channel.attr(ATTR_CLIENT_ID).get();
        if (clientId == null) {
            log.warn("尝试授权未注册的通道");
            return;
        }
        
        channel.attr(ATTR_USER_ID).set(userId);
        channel.attr(ATTR_AUTHORIZED).set(true);
        
        // 记录用户与客户端的关系
        userIdClientIdsMap.computeIfAbsent(userId, k -> ConcurrentHashMap.newKeySet()).add(clientId);
        
        // 更新缓存
        ClientInfo clientInfo = cacheService.get(CacheService.CacheType.CONNECTION, clientId);
        if (clientInfo != null) {
            clientInfo.setUserId(userId);
            clientInfo.setAuthorized(true);
            clientInfo.setAuthTime(System.currentTimeMillis());
            cacheService.put(CacheService.CacheType.CONNECTION, clientId, clientInfo);
        }
        
        log.info("客户端[{}]已授权，用户ID: {}", clientId, userId);
    }
    
    /**
     * 注销连接
     * 
     * @param channel 通道
     * @param reason 原因
     */
    public void unregister(Channel channel, String reason) {
        String clientId = channel.attr(ATTR_CLIENT_ID).get();
        if (clientId == null) {
            log.warn("尝试注销未注册的通道");
            return;
        }
        
        String userId = channel.attr(ATTR_USER_ID).get();
        
        // 移除映射关系
        clientIdChannelMap.remove(clientId);
        if (userId != null) {
            Set<String> clientIds = userIdClientIdsMap.get(userId);
            if (clientIds != null) {
                clientIds.remove(clientId);
                if (clientIds.isEmpty()) {
                    userIdClientIdsMap.remove(userId);
                }
            }
        }
        
        // 更新缓存
        ClientInfo clientInfo = cacheService.get(CacheService.CacheType.CONNECTION, clientId);
        if (clientInfo != null) {
            clientInfo.setDisconnectTime(System.currentTimeMillis());
            clientInfo.setDisconnectReason(reason);
            clientInfo.setConnected(false);
            cacheService.put(CacheService.CacheType.CONNECTION, clientId, clientInfo);
        }
        
        channel.close();
        log.info("客户端[{}]已断开连接，原因: {}，当前连接数: {}", clientId, reason, channelGroup.size() - 1);
    }
    
    /**
     * 断开所有连接
     * 
     * @param reason 原因
     */
    public void disconnectAll(String reason) {
        log.info("断开所有连接，原因: {}", reason);
        channelGroup.forEach(channel -> unregister(channel, reason));
        channelGroup.clear();
        clientIdChannelMap.clear();
        userIdClientIdsMap.clear();
    }
    
    /**
     * 发送消息给指定客户端
     * 
     * @param clientId 客户端ID
     * @param message 消息内容
     * @return 是否发送成功
     */
    public boolean sendToClient(String clientId, String message) {
        ChannelId channelId = clientIdChannelMap.get(clientId);
        if (channelId == null) {
            log.warn("客户端[{}]不存在或已断开连接", clientId);
            return false;
        }
        
        Channel channel = channelGroup.find(channelId);
        if (channel == null || !channel.isActive()) {
            log.warn("客户端[{}]通道不存在或已关闭", clientId);
            clientIdChannelMap.remove(clientId);
            return false;
        }
        
        channel.writeAndFlush(new TextWebSocketFrame(message));
        log.debug("向客户端[{}]发送消息: {}", clientId, message);
        return true;
    }
    
    /**
     * 发送消息给指定用户的所有客户端
     * 
     * @param userId 用户ID
     * @param message 消息内容
     * @return 成功发送的客户端数量
     */
    public int sendToUser(String userId, String message) {
        Set<String> clientIds = userIdClientIdsMap.get(userId);
        if (clientIds == null || clientIds.isEmpty()) {
            log.warn("用户[{}]没有在线客户端", userId);
            return 0;
        }
        
        int successCount = 0;
        for (String clientId : clientIds) {
            if (sendToClient(clientId, message)) {
                successCount++;
            }
        }
        
        log.debug("向用户[{}]的{}个客户端中的{}个发送了消息", userId, clientIds.size(), successCount);
        return successCount;
    }
    
    /**
     * 广播消息给所有客户端
     * 
     * @param message 消息内容
     * @return 成功发送的客户端数量
     */
    public int broadcast(String message) {
        int count = channelGroup.size();
        log.info("广播消息给所有客户端({}个): {}", count, message);
        channelGroup.writeAndFlush(new TextWebSocketFrame(message));
        return count;
    }
    
    /**
     * 获取在线客户端数量
     */
    public int getConnectionCount() {
        return channelGroup.size();
    }
    
    /**
     * 获取在线用户数量
     */
    public int getOnlineUserCount() {
        return userIdClientIdsMap.size();
    }
    
    /**
     * 获取所有客户端信息
     */
    public Map<String, ClientInfo> getAllClientInfo() {
        return clientIdChannelMap.keySet().stream()
                .map(clientId -> cacheService.get(CacheService.CacheType.CONNECTION, clientId))
                .filter(Objects::nonNull)
                .collect(Collectors.toMap(ClientInfo::getClientId, Function.identity()));
    }
    
    /**
     * 获取指定客户端信息
     * 
     * @param clientId 客户端ID
     */
    public ClientInfo getClientInfo(String clientId) {
        return cacheService.get(CacheService.CacheType.CONNECTION, clientId);
    }
    
    /**
     * 获取用户的所有客户端ID
     * 
     * @param userId 用户ID
     */
    public Set<String> getUserClientIds(String userId) {
        return userIdClientIdsMap.getOrDefault(userId, Set.of());
    }
    
    /**
     * 获取所有客户端ID
     */
    public Set<String> getAllClientIds() {
        return new HashSet<>(clientIdChannelMap.keySet());
    }
} 